package com.synchrodata.job;

import java.sql.Connection;
import java.sql.Timestamp;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.log4j.Logger;

import com.synchrodata.util.DataBaseUtil;

public class SynchroDataJob {

	private static Logger logger = Logger.getLogger(SynchroDataJob.class);

	private volatile static ConcurrentHashMap<String,Timestamp> lastMap=new ConcurrentHashMap<String,Timestamp>();

	//用于存储表的列信息
	private static ConcurrentHashMap<String,List<String>> tableMap = new ConcurrentHashMap<String,List<String>>();
	//用于存储表的主键列信息
    private static ConcurrentHashMap<String,String> tableFieldMap = new ConcurrentHashMap<String,String>();

    public static void main(String[] args) {
		new SynchroDataJob().run();
	}


	public  synchronized void run(){
		long start = System.currentTimeMillis();
		System.out.println("开始同步数据");
		logger.info("开始同步数据");
		Connection oneConn = DataBaseUtil.getOneConnection();
		Connection twoConn = DataBaseUtil.getTwoConnection();
		//同步数据 将oneConn同步到twoConn
		try {
			synchroData(oneConn,twoConn,"oneConn");
		} catch (Exception e) {
			e.printStackTrace();
			logger.error("同步数据异常oneConn，错误："+e.getMessage());
		}
		//同步数据 将twoConn同步到oneConn
		try {
			synchroData(twoConn,oneConn,"twoConn");
		} catch (Exception e) {
			e.printStackTrace();
			logger.error("同步数据异常twoConn，错误："+e.getMessage());
		}
		//执行到最后才可以把连接关闭
		DataBaseUtil.close(oneConn,twoConn);
		long end = System.currentTimeMillis();
		System.out.println("同步数据完成,耗时："+(end-start)+"ms");
		logger.info("同步数据完成,耗时：{}"+(end-start)+"ms");
	}

	/**
	 * 同步数据
	 * @param readConn 读取的数据源
	 * @param writeConn 写入的数据源
	 */
	public void synchroData(Connection readConn,Connection writeConn,String numConn) throws Exception{

		Timestamp maxTime=null;
		List<Map<String, Object>>  list = null;

		Timestamp lastMaxTime=lastMap.get(numConn);
		if(lastMaxTime!=null){
			String maxDateSql = "select max(record_time) record_time from (select * from sys_record where record_time >?  order by record_time) re where rownum <= 5000 ";
			List<Object> paras = new ArrayList<Object>();
			paras.add(lastMaxTime);
			list = DataBaseUtil.query(readConn, maxDateSql, paras);
		}else{
			String maxDateSql = "select max(record_time) record_time from (select * from sys_record order by record_time) re where rownum <= 5000 ";
			list = DataBaseUtil.query(readConn, maxDateSql, null);
		}

		if(list!=null && list.size()>0){
			if(list.get(0).get("record_time")!=null){
				maxTime = (Timestamp)list.get(0).get("record_time");
				lastMap.put(numConn, maxTime);
			}else{
				return;
			}
		}else{
			return;
		}

		//查询中间表数据
		String sql = "select * from sys_record where record_time <=? order by record_state,record_time ";
		List<Object> paras = new ArrayList<Object>();
		paras.add(maxTime);

		List<Map<String, Object>>  recordList = DataBaseUtil.query(readConn, sql, paras);
		if(recordList!=null && recordList.size()>0){

			//用于保存删除记录的sql
			List<String> deleteList = new ArrayList<String>();
			//用于保存新增记录的数据Map
			LinkedHashMap<String,List<Map<String, Object>>> insertDataMap = new LinkedHashMap<String,List<Map<String, Object>>>();
			//用于保存修改记录的数据Map
			LinkedHashMap<String,List<Map<String, Object>>> updateDataMap = new LinkedHashMap<String,List<Map<String, Object>>>();

			for(Map<String, Object> recordMap:recordList){
				String record_state = (String)recordMap.get("record_state");
				String table_name = (String)recordMap.get("table_name");
				String table_pkid = (String)recordMap.get("table_pkid");
				String field_name = (String)recordMap.get("field_name");
				//保存表的主键名
				tableFieldMap.put(table_name, field_name);
				//新增操作
				if(record_state.equals("1") ){
					//获取数据
					Map<String, Object> dataMap = getData(readConn,table_name,field_name,table_pkid);
					if(dataMap!=null){
						List<Map<String, Object>> insertDataList = insertDataMap.get(table_name);
						if(insertDataList==null){
							insertDataList = new ArrayList<Map<String, Object>>();
						}
						insertDataList.add(dataMap);
						insertDataMap.put(table_name, insertDataList);
					}
				}else if(record_state.equals("2")){ //修改操作
					//获取数据
					Map<String, Object> dataMap = getData(readConn,table_name,field_name,table_pkid);
					if(dataMap!=null){
						List<Map<String, Object>> updateDataList = updateDataMap.get(table_name);
						if(updateDataList==null){
							updateDataList = new ArrayList<Map<String, Object>>();
						}
						updateDataList.add(dataMap);
						updateDataMap.put(table_name, updateDataList);
					}
				}else if(record_state.equals("3")){ //删除操作
					String deleteSql="delete "+table_name+" where "+field_name+"='"+table_pkid+"' ";
					deleteList.add(deleteSql);
				}
			}

			//执行批处理新增
			if (!insertDataMap.isEmpty()) {
				for (Map.Entry<String, List<Map<String, Object>>> entry : insertDataMap.entrySet()) {
					//获取新增sql
					String insertSql = getInsertSql(readConn,entry.getKey());
					//获取新增的数据
					List<Map<String, Object>> insertDataList = entry.getValue();
					//获取表的列集合
					List<String> colList = getColumnList(readConn,entry.getKey());

					String pkCol = tableFieldMap.get(entry.getKey());
					Map<String, Object> recordMap =new HashMap<String, Object>();
					recordMap.put("table_name", entry.getKey());
					recordMap.put("record_state", "1");
					recordMap.put("record_time", maxTime);

					//执行插入
					DataBaseUtil.updateData(writeConn,insertSql,colList,insertDataList,recordMap,pkCol,readConn);
				}
			}

			//执行批处理修改
			if (!updateDataMap.isEmpty()) {
				for (Map.Entry<String, List<Map<String, Object>>> entry : updateDataMap.entrySet()) {
					//获取修改sql
					String updateSql = getUpdateSql(readConn,entry.getKey());
					//获取修改的数据
					List<Map<String, Object>> updateDataList = entry.getValue();
					//获取表的列集合
					List<String> colList = getColumnList(readConn,entry.getKey());

					List<String> newColList = new ArrayList<String>();
					newColList.addAll(colList);

					String pkCol = tableFieldMap.get(entry.getKey());

					//给集合插入最后主键列名,因为修改sql最后需要 =pkid
					newColList.add(pkCol);

					Map<String, Object> recordMap =new HashMap<String, Object>();
					recordMap.put("table_name", entry.getKey());
					recordMap.put("record_state", "2");
					recordMap.put("record_time", maxTime);

					//执行批处理 插入
					DataBaseUtil.updateData(writeConn,updateSql,newColList,updateDataList,recordMap,pkCol,readConn);
				}
			}
			//执行批处理删除
			DataBaseUtil.sqlBatch(writeConn, deleteList);

			//删除数据
			String deleteSql = "delete from sys_record where record_time <=? and record_state='3' ";
			DataBaseUtil.execute(readConn, deleteSql,paras);
		}
	}

	/**
	 * 获取表的列信息
	 * @param conn
	 * @param tableName
	 * @return
	 */
	public List<String> getColumnList(Connection conn, String tableName){
		List<String> colList = tableMap.get(tableName);
		if(colList==null){
			 colList = DataBaseUtil.getColumnList(conn, tableName);
			 tableMap.put(tableName, colList);
		}
		return colList;
	}

	/**
	 * 获取数据
	 * @param conn 链接
	 * @param tableName 表名
	 * @param fieldName 主键列名
	 * @param tablePkid 主键值
	 * @return
	 */
	public Map<String, Object> getData(Connection conn,String tableName,String fieldName,String tablePkid){
		String dataSql = "select * from "+tableName+" where "+fieldName+"='"+tablePkid+"' ";
		List<Map<String, Object>>  dataList = DataBaseUtil.query(conn, dataSql, null);
		if(dataList!=null && dataList.size()>0){
			return dataList.get(0);
		}else{
			return null;
		}
	}

	/**
	 * 根据表名获取insertSql
	 * @param conn
	 * @param tableName
	 * @return
	 */
	public String getInsertSql(Connection conn,String tableName){
		//获取表的列信息
		List<String> colList = getColumnList(conn, tableName);
		if(colList==null){
			return null;
		}
		//拼接insertSql
		StringBuilder insertSql = new StringBuilder();
		insertSql.append("insert into "+tableName+" (");
		StringBuilder colSql = new StringBuilder();
		StringBuilder valSql = new StringBuilder();
		for(String colName:colList){
			colSql.append(colName+",");
			valSql.append("?,");
		}
		insertSql.append(colSql.substring(0, colSql.length()-1)+") values(");
		insertSql.append(valSql.substring(0, valSql.length()-1)+") ");
		return insertSql.toString();
	}

	/**
	 * 根据表名获取UpdateSql
	 * @param conn
	 * @param tableName
	 * @return
	 */
	public String getUpdateSql(Connection conn,String tableName){
		//获取表的列信息
		List<String> colList = getColumnList(conn, tableName);
		if(colList==null){
			return null;
		}
		//拼接updateSql
		StringBuilder updateSql = new StringBuilder();
		updateSql.append("update "+tableName+" set ");
		StringBuilder colSql = new StringBuilder();
		for(String colName:colList){
			colSql.append(colName+"=?,");
		}
		updateSql.append(colSql.substring(0, colSql.length()-1)+" where "+tableFieldMap.get(tableName)+"=? ");
		return updateSql.toString();
	}
}
